package com.mqz.rocketmq.producer.component;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * 版权所有  copyright© 蒙大拿
 *
 * @author mqz
 * @date 2022/7/28
 * @about https://www.gitee.com/DemoMeng
 * @description
 */
@Component
@Slf4j
public class Producer<T> {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String msg){
        rocketMQTemplate.convertAndSend(topic, msg);
    }


    /**
     * 同步发送消息
     * @param message
     */
    public void sendSyncMessage(Message message){
        log.info("同步发送消息：{}",message);
        SendResult sendResult = rocketMQTemplate.syncSend("broker-a", message);
        log.info("同步发送消息结果：{}",sendResult);
    }

    /**
     * 异步发送消息
     * @param message
     */
    public void sendAsyncMessage(Message message){
        log.info("异步发送消息：{}",message);
        rocketMQTemplate.asyncSend("broker-a", message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送消息结果：{}",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("异步发送消息异常：{}",throwable);
            }
        });
    }

    /**
     * 单向发送消息
     * @param message
     */
    public void sendOnewayMessage(Message message){
        log.info("单向发送消息：{}",message);
        rocketMQTemplate.sendOneWay("broker-a", message);
    }


    /**
     * 定时投递消息
     * @param message
     */
    public void delayTimeSendMessage(org.springframework.messaging.Message<Map<String,Object>> message,long second){
        rocketMQTemplate.syncSend("broker-a",message,second,2);
    }

    /**
     * 消息过滤tag
     * @param message
     */
    public void filterSendTag(String tag, org.springframework.messaging.Message<String> message){
        rocketMQTemplate.sendOneWay("mengqizhang:"+tag,message);
    }

    /**
     * 消息过滤SQL92
     * @param message
     */
    public void filterSendSQL92(org.springframework.messaging.Message<String> message,Map<String,Object> headers){
        rocketMQTemplate.convertAndSend("mengqizhang-sql-topic",message,headers);
    }



}
